package hk.ust.cse.comp4981;

import hk.ust.cse.comp4981.EventDetection;
import hk.ust.cse.comp4981.EventDetection.Time;


import java.io.IOException;
import java.text.Format;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import org.joda.time.DateTime;
import twitter4j.FilterQuery;
import twitter4j.GeoLocation;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterException;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;

public class StreamingAPI{
	static TwitterStream twitterstream = new TwitterStreamFactory().getInstance();
	static FilterQuery tweetFilterQuery = new FilterQuery();
	static ArrayList<MyTweet> sanTweets = new ArrayList<MyTweet>();
	static ArrayList<MyTweet> nyTweets = new ArrayList<MyTweet>();
	static int s, n, total, count = 0;
	static jdbcmysql test = new jdbcmysql(); 
	static Format formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	static double latitude;
	static double longitude;
	static GeoLocation[][] boundBox;
	static double[] boundLong = new double[4];
	static double[] boundLat = new double[4];
	static int userid;
	static long test_time;
	static long endtime_set;
	static long diff_set;
	static double diff_time;	
	static String d;
	private final static Object lock = new Object();
	
	public StreamingAPI() {
		super();
	}
	
	public static void assignDefaultListener(int sample){
	    StatusListener listener = new StatusListener(){
	        public void onStatus(Status status) {
	        	if(status.getGeoLocation() != null){
	        		latitude = status.getGeoLocation().getLatitude();
	        		longitude = status.getGeoLocation().getLongitude();
	        	}
	        	else{
	        		latitude = 0.0;
	        		longitude = 0.0;
	        	}
	        	
	        	if(status.getPlace().getBoundingBoxCoordinates() != null){
	        		boundBox = status.getPlace().getBoundingBoxCoordinates();
	        		count = 0;
	        		for(GeoLocation[] geoRow:boundBox){
	        			for(GeoLocation geoCol:geoRow){
	        				boundLong[count] = geoCol.getLongitude();
	    	        		boundLat[count] = geoCol.getLatitude();
	    	        		count++;
	        			}
	        		}
	        	}
	        	else{
	        		for(int h = 0; h < 4; h++){
	        			boundLong[h] = 0.0;
	        			boundLat[h] = 0.0;
	        		}
	        	}
	        	
	        	
	        	//store San Francisco's tweets
	        	if((longitude >= -122.75 && longitude <= -121.75 && latitude >= 36.8 && latitude <= 37.8) || 
	        			(boundLong[0] >= -122.75 && boundLong[0] <= -121.75 && boundLat[0] >= 36.8 && boundLat[0] <= 37.8) || 
	        			(boundLong[1] >= -122.75 && boundLong[1] <= -121.75 && boundLat[1] >= 36.8 && boundLat[1] <= 37.8) ||
	        			(boundLong[2] >= -122.75 && boundLong[2] <= -121.75 && boundLat[2] >= 36.8 && boundLat[2] <= 37.8) ||
	        			(boundLong[3] >= -122.75 && boundLong[3] <= -121.75 && boundLat[3] >= 36.8 && boundLat[3] <= 37.8)){
	        		sanTweets.add(new MyTweet(status));
	        		System.out.println();
	        	
	        		userid = test.insertTable_user(status.getUser().getScreenName(), status.getUser().getFavouritesCount(), status.getUser().getFollowersCount(), status.getUser().getFriendsCount(), 0);
	        		d = formatter.format(status.getCreatedAt());
	        		test.insertTable_sf(userid, 2, d, status.getText());
	        		s++; 
	        		total++;
	        		//System.out.println("s: " + s + "total: " + total); //for testing
	        		//data mining
		        	if (s==1000){  
		        		try {
							Twokenize.DoSanTokenization(); 
						} catch (IOException e1) {
							// TODO Auto-generated catch block
							e1.printStackTrace();
						}
		        		
		        		try {
							Ngrams.generateSanNgrams();
						} catch (IOException e1) {
							// TODO Auto-generated catch block
							e1.printStackTrace();
						}
		        		
		        		try {
							tfidf.generateSANTfidf();
						} catch (IOException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
		        		
		        		SanClusters.cluster();
		        		s=0;
		        		// remove all the contents of arraylist
		        	}
	        	}

	        	
	        	//store New York's tweets
	        	if((longitude >= -74 && longitude <= -73 && latitude >= 40 && latitude <= 41)||
	        			(boundLong[0] >= -74 && boundLong[0] <= -73 && boundLat[0] >= 40 && boundLat[0] <= 41) ||
	        			(boundLong[1] >= -74 && boundLong[1] <= -73 && boundLat[1] >= 40 && boundLat[1] <= 41) ||
	        			(boundLong[2] >= -74 && boundLong[2] <= -73 && boundLat[2] >= 40 && boundLat[2] <= 41) ||
	        			(boundLong[3] >= -74 && boundLong[3] <= -73 && boundLat[3] >= 40 && boundLat[3] <= 41) ){
	        		// nyContent = status.getCreatedAt() + " : " + status.getText() + " latitude: " + latitude + " longitude: " + longitude;
	        		nyTweets.add(new MyTweet(status));
	        		
	        		//insert the user information into the 'twitter_user' table and get the newest userid
	        		userid = test.insertTable_user(status.getUser().getScreenName(), status.getUser().getFavouritesCount(), status.getUser().getFollowersCount(), status.getUser().getFriendsCount(), 0);
	        		d = formatter.format(status.getCreatedAt());
	        		//insert the tweet information and new userid into 'ny_tweet' table
	        		test.insertTable_ny(userid, 1, d, status.getText());
	        		n++;//counting no. of tweets
	        		total++;
	        		//System.out.println("n: " + n + "total: " + total); //for testing
	        		
	        		//data mining
		        	if (n==1000){  
		        		try {
							Twokenize.DoNyTokenization(); 
						} catch (IOException e1) {
							// TODO Auto-generated catch block
							e1.printStackTrace();
						}
		        		
		        		try {
							Ngrams.generateNyNgrams();
						} catch (IOException e1) {
							// TODO Auto-generated catch block
							e1.printStackTrace();
						}
		        		
		        		try {
							tfidf.generateNYTfidf();
						} catch (IOException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
		        		
		        		NyClusters.cluster();
		        		
		        		n=0;
		        		// remove all the contents of arraylist 
		        	}
	        	}
	        	
	        	endtime_set = System.currentTimeMillis();
	        	diff_set = (endtime_set -  Time.starttime_set);
	        	//diff_time = diff_set / 1000.0 ; //seconds
	        	//diff_time = diff_set / 1000.0 / 60.0; //minutes
	        	diff_time = diff_set / 1000.0 / 60.0 / 60.0; //hours
	        	
	        	//use no. of tweets as ending condition
	        	/*if (total >= sample){

	        		synchronized (lock) {
		                  lock.notify();
		                }
		                System.out.println("unlocked");
		                System.out.println("Time cost: " + diff_time + " (in mins)");
	        	}*/
	        	
	        	//use duration as ending condition
	        	if (diff_time >= sample){
	        		synchronized (lock) {
		                  lock.notify();
		                }
		                System.out.println("unlocked");
		                System.out.println("Time cost: " + diff_time + " (in mins)");
	        	}
	        }
	        public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {}
	        public void onTrackLimitationNotice(int numberOfLimitedStatuses) {}
	        public void onException(Exception ex) {
	            ex.printStackTrace();
	        }
			@Override
			public void onScrubGeo(long arg0, long arg1) {
				// TODO Auto-generated method stub
				
			}
			@Override
			public void onStallWarning(StallWarning arg0) {
				// TODO Auto-generated method stub
				
			}
	    };
	    
	    twitterstream.addListener(listener);
	    // sample() method internally creates a thread which manipulates TwitterStream and calls these adequate listener methods continuously.
	    twitterstream.filter(tweetFilterQuery);
	    
	}
	
	public static void openfile(int size) throws TwitterException, IOException{
    	DateTime start_time = new DateTime();
		Time.starttime_set = start_time.getMillis();
    	tweetFilterQuery.locations(new double[][]{new double[]{-122.75,36.8},{-121.75,37.8}, new double[]{-74, 40},{-73, 41}});//collect data from San Francisco OR New York City
    	//tweetFilterQuery.locations(new double[][]{new double[]{26.3,1.16}, new double[]{169.40,77.43}}); //collect data from Asia
    	//tweetFilterQuery.locations(new double[][]{new double[]{-180,-90}, new double[]{180,90}}); //collect data from the whole world
    	tweetFilterQuery.language(new String[]{"en"});
//    	test_tf(); //for testing tf-idf
    	consumeTweets(size);

    	//EventDetection.test(sanTweets, nyTweets); //testing data crawler
	    //EventDetection_test.end();
	}
	
	public static void consumeTweets(int size){
		assignDefaultListener(size);
		try {
	        synchronized (lock) {
	          lock.wait();
	        }
	      } catch (InterruptedException e) {
	        // TODO Auto-generated catch block
	        e.printStackTrace();
	      }
		twitterstream.cleanUp();
		System.out.println("Cleaned");
		twitterstream.shutdown();
		System.out.println("Shut down");
		

		//EventDetection.test(sanTweets, nyTweets); //testing data crawler
	    EventDetection.end();

	}
/*	
	//for testing tf-idf
	public static void test_tf(){
			test.SelectTable_tf();
		
    		try {
				Twokenize.DoNyTokenization(); 
			} catch (IOException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
    		
    		try {
				Ngrams.generateNyNgrams();
			} catch (IOException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
    		
    		try {
				tfidf.generateTfidf();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
    		
	}
*/
	

	
	
}